import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.flink.api.common.operators.Order;




public class SortPartition
{


    public static void main(String[] args) throws Exception
    {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).append("/dataset_api/Java").toString();
        DataSet<Tuple2<String,Integer>> in = env.readCsvFile("file:///"+rootPath+"/"+"cog1.csv").types(String.class,Integer.class);

        System.out.println("--------------------------in------------------------------");
        in.print();
// Locally sort partitions in ascending order on the second String field and
// in descending order on the first String field.
// Apply a MapPartition transformation on the sorted partitions.
        DataSet<Tuple2<String, String>> out = in.sortPartition(1, Order.ASCENDING)
                .sortPartition(0, Order.DESCENDING)
                .mapPartition(new MapPartitionFunction<Tuple2<String, Integer>, Tuple2<String, String>>()
                {
                    @Override
                    public void mapPartition(Iterable<Tuple2<String, Integer>> iterable, Collector<Tuple2<String, String>> out) throws Exception
                    {

                        String key="keydefault";
                        String sum="sumdefault";

                        for(Tuple2<String,Integer>cur:iterable)
                        {
                            key=key+cur.f0;
                            sum=sum+cur.f1.toString();


                        }
                        out.collect(new Tuple2<String, String>(key,sum));
                    }
                });


        System.out.println("--------------------------out的数据------------------------------");
        out.print();



    }

}
